/*
 * XDevice RPC Server Implementation
 * Provides Unix Domain Socket interface for monitoring Raft cluster status
 */

#include <xdevice/rpc.h>
#include <xdevice/xdevice.h>
#include <xdevice/raft.h>

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <errno.h>
#include <time.h>

static int g_server_fd = -1;
static pthread_t g_server_thread;
static bool g_server_running = false;
static xdevice_context_t* g_xdevice_context = NULL;
static pthread_mutex_t g_context_lock = PTHREAD_MUTEX_INITIALIZER;

/* Utility functions */
static int send_response(int client_fd, uint32_t request_id, uint32_t result, 
                        const void* data, uint32_t data_length) {
    xdevice_rpc_message_t* response = malloc(sizeof(xdevice_rpc_message_t) + data_length);
    if (!response) {
        return -1;
    }
    
    response->type = result;
    response->request_id = request_id;
    response->data_length = data_length;
    
    if (data && data_length > 0) {
        memcpy(response->data, data, data_length);
    }
    
    ssize_t total_size = sizeof(xdevice_rpc_message_t) + data_length;
    ssize_t sent = send(client_fd, response, total_size, 0);
    
    free(response);
    return (sent == total_size) ? 0 : -1;
}

/* RPC command handlers */
static int handle_get_status(int client_fd, uint32_t request_id) {
    xdevice_rpc_node_status_t status = {0};
    
    pthread_mutex_lock(&g_context_lock);
    
    if (g_xdevice_context) {
        // Get basic node info using accessor functions
        status.node_id = xdevice_get_node_id(g_xdevice_context);
        status.is_healthy = true;
        
        // Try to get cluster info
        xdevice_cluster_info_t cluster_info;
        if (xdevice_get_cluster_info(g_xdevice_context, &cluster_info) == XDEVICE_CORE_OK) {
            status.leader_id = cluster_info.leader_id;
            status.current_term = cluster_info.current_term;
            status.commit_index = cluster_info.commit_index;
            status.is_healthy = cluster_info.is_healthy;
            
            // Determine node state
            if (status.leader_id == status.node_id) {
                status.state = 2;  // leader
            } else if (status.leader_id > 0) {
                status.state = 0;  // follower
            } else {
                status.state = 1;  // candidate
            }
        }
        
        // Set some default values
        status.election_timeout_ms = 300;
        status.heartbeat_interval_ms = 50;
        status.last_heartbeat_time = time(NULL) * 1000;
    }
    
    pthread_mutex_unlock(&g_context_lock);
    
    return send_response(client_fd, request_id, XDEVICE_RPC_OK, &status, sizeof(status));
}

static int handle_get_cluster_info(int client_fd, uint32_t request_id) {
    xdevice_rpc_cluster_status_t cluster_status = {0};
    
    pthread_mutex_lock(&g_context_lock);
    
    if (g_xdevice_context) {
        xdevice_cluster_info_t cluster_info;
        if (xdevice_get_cluster_info(g_xdevice_context, &cluster_info) == XDEVICE_CORE_OK) {
            cluster_status.node_count = cluster_info.node_count;
            cluster_status.active_nodes = cluster_info.node_count;
            cluster_status.leader_id = cluster_info.leader_id;
            cluster_status.cluster_term = cluster_info.current_term;
            cluster_status.cluster_healthy = cluster_info.is_healthy;
            
            // Fill in node info (simplified)
            for (int i = 0; i < cluster_info.node_count && i < 16; i++) {
                xdevice_rpc_node_status_t* node = &cluster_status.nodes[i];
                node->node_id = i + 1;
                node->current_term = cluster_info.current_term;
                node->commit_index = cluster_info.commit_index;
                node->leader_id = cluster_info.leader_id;
                node->is_healthy = true;
                
                if (node->node_id == cluster_info.leader_id) {
                    node->state = 2;  // leader
                } else {
                    node->state = 0;  // follower
                }
            }
        }
    }
    
    pthread_mutex_unlock(&g_context_lock);
    
    return send_response(client_fd, request_id, XDEVICE_RPC_OK, &cluster_status, sizeof(cluster_status));
}

static int handle_get_log_info(int client_fd, uint32_t request_id) {
    xdevice_rpc_log_info_t log_info = {0};
    
    pthread_mutex_lock(&g_context_lock);
    
    if (g_xdevice_context) {
        // Get basic log info
        log_info.start_index = 1;
        log_info.end_index = 10;  // Example
        log_info.commit_index = 5;  // Example
        log_info.entry_count = 3;   // Example
        
        // Fill in some sample log entries
        for (uint32_t i = 0; i < log_info.entry_count && i < 10; i++) {
            log_info.entries[i].index = i + 1;
            log_info.entries[i].term = 1;
            log_info.entries[i].type = 1;  // Application entry
            log_info.entries[i].data_size = 64;
            snprintf(log_info.entries[i].data_preview, sizeof(log_info.entries[i].data_preview),
                    "Log entry %u data...", i + 1);
        }
    }
    
    pthread_mutex_unlock(&g_context_lock);
    
    return send_response(client_fd, request_id, XDEVICE_RPC_OK, &log_info, sizeof(log_info));
}

static int handle_get_devices(int client_fd, uint32_t request_id) {
    xdevice_rpc_device_list_t device_list = {0};
    
    pthread_mutex_lock(&g_context_lock);
    
    if (g_xdevice_context) {
        char device_names[XDEVICE_MAX_DEVICES][XDEVICE_MAX_DEVICE_NAME];
        int count = xdevice_list_devices(g_xdevice_context, device_names, XDEVICE_MAX_DEVICES);
        
        device_list.device_count = (count > 32) ? 32 : count;
        
        for (int i = 0; i < device_list.device_count; i++) {
            strncpy(device_list.devices[i].name, device_names[i], 
                   sizeof(device_list.devices[i].name) - 1);
            device_list.devices[i].size = 1024 * 1024;  // Default size
            device_list.devices[i].state = 1;  // Active
            device_list.devices[i].storage_type = 1;  // Local
            device_list.devices[i].readonly = false;
        }
    }
    
    pthread_mutex_unlock(&g_context_lock);
    
    return send_response(client_fd, request_id, XDEVICE_RPC_OK, &device_list, sizeof(device_list));
}

static int handle_ping(int client_fd, uint32_t request_id) {
    const char* pong = "pong";
    return send_response(client_fd, request_id, XDEVICE_RPC_OK, pong, strlen(pong));
}

/* Client handler */
static void handle_client(int client_fd) {
    xdevice_rpc_message_t message;
    
    // Read message header
    ssize_t received = recv(client_fd, &message, sizeof(message), 0);
    if (received != sizeof(message)) {
        printf("[RPC] Invalid message from client\n");
        return;
    }
    
    printf("[RPC] Received request: type=%u, id=%u, data_len=%u\n", 
           message.type, message.request_id, message.data_length);
    
    // Handle different message types
    switch (message.type) {
        case XDEVICE_RPC_GET_STATUS:
            handle_get_status(client_fd, message.request_id);
            break;
            
        case XDEVICE_RPC_GET_CLUSTER_INFO:
            handle_get_cluster_info(client_fd, message.request_id);
            break;
            
        case XDEVICE_RPC_GET_RAFT_LOG:
            handle_get_log_info(client_fd, message.request_id);
            break;
            
        case XDEVICE_RPC_GET_DEVICES:
            handle_get_devices(client_fd, message.request_id);
            break;
            
        case XDEVICE_RPC_PING:
            handle_ping(client_fd, message.request_id);
            break;
            
        default:
            send_response(client_fd, message.request_id, XDEVICE_RPC_INVALID_REQUEST, NULL, 0);
            break;
    }
}

/* Server thread */
static void* server_thread(void* arg) {
    const char* socket_path = (const char*)arg;
    
    printf("[RPC] Server started on socket: %s\n", socket_path);
    
    while (g_server_running) {
        struct sockaddr_un client_addr;
        socklen_t client_len = sizeof(client_addr);
        
        int client_fd = accept(g_server_fd, (struct sockaddr*)&client_addr, &client_len);
        if (client_fd < 0) {
            if (g_server_running) {
                perror("[RPC] accept");
            }
            continue;
        }
        
        printf("[RPC] Client connected\n");
        handle_client(client_fd);
        close(client_fd);
        printf("[RPC] Client disconnected\n");
    }
    
    printf("[RPC] Server thread exiting\n");
    return NULL;
}

/* Public API implementation */
int xdevice_rpc_server_start(const char* socket_path) {
    if (g_server_running) {
        return -1;  // Already running
    }
    
    // Create socket
    g_server_fd = socket(AF_UNIX, SOCK_STREAM, 0);
    if (g_server_fd < 0) {
        perror("socket");
        return -1;
    }
    
    // Remove existing socket file
    unlink(socket_path);
    
    // Bind socket
    struct sockaddr_un server_addr;
    memset(&server_addr, 0, sizeof(server_addr));
    server_addr.sun_family = AF_UNIX;
    strncpy(server_addr.sun_path, socket_path, sizeof(server_addr.sun_path) - 1);
    
    if (bind(g_server_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)) < 0) {
        perror("bind");
        close(g_server_fd);
        g_server_fd = -1;
        return -1;
    }
    
    // Listen
    if (listen(g_server_fd, 5) < 0) {
        perror("listen");
        close(g_server_fd);
        g_server_fd = -1;
        return -1;
    }
    
    // Start server thread
    g_server_running = true;
    if (pthread_create(&g_server_thread, NULL, server_thread, (void*)socket_path) != 0) {
        perror("pthread_create");
        g_server_running = false;
        close(g_server_fd);
        g_server_fd = -1;
        return -1;
    }
    
    return 0;
}

void xdevice_rpc_server_stop(void) {
    if (!g_server_running) {
        return;
    }
    
    g_server_running = false;
    
    if (g_server_fd >= 0) {
        close(g_server_fd);
        g_server_fd = -1;
    }
    
    pthread_join(g_server_thread, NULL);
    printf("[RPC] Server stopped\n");
}

int xdevice_rpc_server_set_context(void* xdevice_ctx) {
    pthread_mutex_lock(&g_context_lock);
    g_xdevice_context = (xdevice_context_t*)xdevice_ctx;
    pthread_mutex_unlock(&g_context_lock);
    return 0;
}
